package com.xframe.event.bus.core.msg;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Envelope;
import com.xframe.event.bus.core.annotation.ReceiveMsgArgs;

import java.io.IOException;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
/**
 * 基础的接受器实现
 *
 * @author xufeng
 * @version 1.0
 * @date 2022-09-19
 */
public abstract class BaseMsgReceive<T> implements IMsgProc{

    /**
     * 消息接收器
     */
    protected IReceiveMsg receiveMsg;
    /**
     * 获得泛型参数类型
     */
    protected Class<T> getTClass() {
        Type type = getClass().getGenericSuperclass();
        if(!(type instanceof ParameterizedType)) {
            throw new IllegalStateException("Type must be a parameterized type");
        }
        ParameterizedType parameterizedType = (ParameterizedType)type;
        // 获取泛型的具体类型  这里是单泛型
        Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
        if(null == actualTypeArguments || actualTypeArguments.length<1) {
            throw new IllegalStateException("Number of type arguments must be 1");
        }
        return (Class<T>) actualTypeArguments[0];
    }
    /**
     * 泛型参数类型
     */
    protected Class<?> cls;
    /**
     * 初始化
     * @param receiveMsgArgs 设置参数
     */
    public void init(ReceiveMsgArgs receiveMsgArgs){
        if(getExchange() != null){
            receiveMsgArgs.setExchange(getExchange());
        }
        if(getQueue() != null){
            receiveMsgArgs.setQueue(getQueue());
        }
        if(routingKey() != null){
            receiveMsgArgs.setRoutingKey(routingKey());
        }
        receiveMsg = ReceiveMsgFactory.create(receiveMsgArgs,this);
        cls = getTClass();
        receiveMsg.init();
    }
    /**
     * 交换机名称 子类可以Override改方法，优先级比注解高
     */
    public String getExchange(){
        return null;
    };
    /**
     * 队列名称,多个使用逗号隔开
     * 子类可以Override改方法，优先级比注解高
     */
    public String getQueue(){
        return null;
    };
    /**
     * 路由KEY，多个使用逗号隔开
     * 子类可以Override改方法，优先级比注解高
     */
    public String routingKey(){
        return null;
    };

    /**
     * 消息接收入口函数
     * @param consumerTag 用户自定义的tag
     * @param envelope 消息的环境上下文对象,可以获得队列名称,路由key等参数
     * @param properties 附加属性
     * @param msg 消息
     */
    @Override
    public void proc(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, String msg) {
        T msgObj = getMsgObj(msg);
        if(msgObj != null) {
            procMsg(consumerTag, envelope, properties, msgObj, msg);
        }else {
            // 发送消息到死信队列
        }
    }

    /**
     * 将消息转换为指定的对象
     * @param msg 消息
     *
     * @return T类型的对象
     */
    protected T getMsgObj(String msg){
        try {
            if (cls.isAssignableFrom(String.class)) {
                return (T) msg;
            }
            JSONObject jsonObject = JSON.parseObject(msg);
            T msgObj = (T) JSON.toJavaObject(jsonObject, cls);
            return msgObj;
        }catch (Exception e){
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 手动ACK
     * @param deliveryTag 消息的环境上下文对象ID
     */
    protected void ack(long deliveryTag) throws IOException {
        this.receiveMsg.getCurChannel().basicAck(deliveryTag, false);
    }
    /**
     * 子类处理消息的方法
     * @param consumerTag 用户自定义的tag
     * @param envelope 消息的环境上下文对象,可以获得队列名称,路由key等参数
     * @param properties 附加属性
     * @param msgObj 消息对象
     * @param msg 原始的消息
     */
    public abstract void procMsg(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, T msgObj,String msg);
}
